package net.bwie.realtime.jtp.dws.douyin.log.functions;

import net.bwie.realtime.jtp.dws.douyin.log.bean.EventLog6;
import net.bwie.realtime.jtp.dws.douyin.log.bean.InteractionTrendMetric6;
import net.bwie.realtime.jtp.dws.douyin.log.bean.UserFanRelation6;
import net.bwie.realtime.jtp.utils.DateTimeUtil;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.Iterator;

public class DouYinTrendWindowFunction6 implements WindowFunction<
        Tuple2<EventLog6, UserFanRelation6>,  // 输入数据（日志+粉丝关系）
        InteractionTrendMetric6,                 // 输出指标
        Long,                                   // 分组键（直播间ID）
        TimeWindow> {                           // 时间窗口

    @Override
    public void apply(
            Long liveRoomId,                    // 分组键：直播间ID
            TimeWindow window,                  // 窗口信息
            Iterable<Tuple2<EventLog6, UserFanRelation6>> input,  // 窗口内数据
            Collector<InteractionTrendMetric6> out) throws Exception {

        // 1. 初始化统计变量
        long newCommentCount = 0;  // 新增评论数
        long newFanCount = 0;      // 新增粉丝数
        Long anchorId = null;      // 主播ID（单直播间内唯一）

        // 2. 遍历窗口内数据，分类统计
        Iterator<Tuple2<EventLog6, UserFanRelation6>> iterator = input.iterator();
        while (iterator.hasNext()) {
            Tuple2<EventLog6, UserFanRelation6> tuple = iterator.next();
            EventLog6 eventLog = tuple.f0;
            UserFanRelation6 fanRelation = tuple.f1;

            // 提取主播ID（单直播间内所有数据主播ID一致）
            if (anchorId == null && eventLog != null) {
                anchorId = eventLog.getAnchorId();
            } else if (anchorId == null && fanRelation != null) {
                anchorId = fanRelation.getAnchorId();
            }

            // 统计新增评论数（过滤日志表中event_type='评论'的记录）
            if (eventLog != null && "评论".equals(eventLog.getEventType())) {
                newCommentCount++;
            }

            // 统计新增粉丝数（过滤粉丝表中is_new='yes'的记录）
            if (fanRelation != null && "yes".equals(fanRelation.getIsNew())) {
                newFanCount++;
            }
        }

        // 3. 格式化窗口时间（转换为yyyy-MM-dd HH:mm:ss）
        long windowStart = window.getStart();
        long windowEnd = window.getEnd();
        String windowStartTime = DateTimeUtil.convertLongToString(windowStart, "yyyy-MM-dd HH:mm:ss");
        String windowEndTime = DateTimeUtil.convertLongToString(windowEnd, "yyyy-MM-dd HH:mm:ss");
        String curDate = windowStartTime.substring(0, 10);  // 提取日期（yyyy-MM-dd）

        // 4. 封装指标结果并输出
        InteractionTrendMetric6 metric = new InteractionTrendMetric6(
                windowStartTime,
                windowEndTime,
                curDate,
                liveRoomId,
                anchorId,
                newCommentCount,
                newFanCount
        );
        out.collect(metric);
    }
}
